package cn.hnzzs.canal;


import cn.hnzzs.dmo.ColumnBean;
import cn.hnzzs.dmo.ExecBean;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author Created by Alan on 2017/4/10.
 */
public class CanalClient {

    private static Logger logger = Logger.getLogger(CanalClient.class);
    private static Properties prop = new Properties();

    static {
        try {
            prop.load(CanalClient.class.getResourceAsStream("/canalClient.properties"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void execEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChange;

            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                logger.error("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);

            }

            CanalEntry.EventType eventType = rowChange.getEventType();
            String tableName = entry.getHeader().getTableName();

            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.DELETE) {
                    deleteData(rowData.getBeforeColumnsList(), tableName);
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    insertData(rowData.getAfterColumnsList(), tableName);
                } else {
                    updateData(rowData.getAfterColumnsList(), tableName);
                }
            }
        }
    }

    private static ExecBean insertSql(List<CanalEntry.Column> columns, String tableName) {
        StringBuilder execSql = new StringBuilder("INSERT INTO ");
        StringBuilder placeHolder = new StringBuilder();
        ExecBean execBean = new ExecBean();
        List<ColumnBean> columnBeans = new ArrayList<>();
        execSql.append(tableName).append(" (");

        for (CanalEntry.Column column : columns) {
            ColumnBean columnBean = new ColumnBean();
            execSql.append(column.getName());
            execSql.append(",");

            placeHolder.append("?");
            placeHolder.append(",");

            columnBean.setColumnName(column.getName());
            columnBean.setColumnValue(column.getValue());
            columnBean.setJdbcType(column.getSqlType());
            columnBean.setIsNull(column.getIsNull());
            columnBean.setColumnUpdate(column.getUpdated());
            columnBeans.add(columnBean);
        }
        execSql.deleteCharAt(execSql.length() - 1);
        execSql.append(") VALUES(");
        placeHolder.deleteCharAt(placeHolder.length() - 1);
        execSql.append(placeHolder);
        execSql.append(");");

        execBean.setExecSql(execSql.toString());
        execBean.setColumnBeanList(columnBeans);

        return execBean;
    }

    private static ExecBean updateSql(List<CanalEntry.Column> columns, String tableName) {

        ExecBean execBean = new ExecBean();
        List<ColumnBean> columnBeans = new ArrayList<>();

        StringBuilder execSql = new StringBuilder("UPDATE ");

        execSql.append(tableName);
        execSql.append(" SET ");

        ColumnBean columnBeanIdKey = new ColumnBean();

        for (CanalEntry.Column column : columns) {
            ColumnBean columnBean = new ColumnBean();
            if (column.getUpdated()) {
                execSql.append(column.getName());
                execSql.append(" =  ?,");

                columnBean.setColumnName(column.getName());
                columnBean.setColumnValue(column.getValue());
                columnBean.setJdbcType(column.getSqlType());
                columnBean.setIsNull(column.getIsNull());
                columnBean.setColumnUpdate(column.getUpdated());
                columnBeans.add(columnBean);
            }

            if (column.getIsKey()) {
                columnBeanIdKey.setColumnName(column.getName());
                columnBeanIdKey.setColumnValue(column.getValue());
                columnBeanIdKey.setIsNull(column.getIsNull());
                columnBeanIdKey.setJdbcType(column.getSqlType());
            }
        }

        columnBeans.add(columnBeanIdKey);

        execSql.deleteCharAt(execSql.length() - 1);
        execSql.append(" WHERE ");
        execSql.append(columnBeanIdKey.getColumnName());
        execSql.append(" = ?;");
        execBean.setExecSql(execSql.toString());
        execBean.setColumnBeanList(columnBeans);
        return execBean;

    }

    private static ExecBean deleteSql(List<CanalEntry.Column> columns, String tableName) {
        ExecBean execBean = new ExecBean();
        List<ColumnBean> columnBeanList = new ArrayList<>();

        ColumnBean columnBeanIdKey = new ColumnBean();
        StringBuilder execSql = new StringBuilder("DELETE FROM ");
        execSql.append(tableName);
        execSql.append(" WHERE ");

        for (CanalEntry.Column column : columns) {

            if (column.getIsKey()) {
                columnBeanIdKey.setColumnName(column.getName());
                columnBeanIdKey.setIsNull(column.getIsNull());
                columnBeanIdKey.setColumnValue(column.getValue());
                columnBeanIdKey.setJdbcType(column.getSqlType());
            }
        }
        execSql.append(columnBeanIdKey.getColumnName());
        execSql.append(" = ?;");
        columnBeanList.add(columnBeanIdKey);
        execBean.setColumnBeanList(columnBeanList);
        execBean.setExecSql(execSql.toString());
        return execBean;
    }


    private static void insertData(List<CanalEntry.Column> columns, String tableName) {

        ExecBean execBean = insertSql(columns, tableName);
        logger.info("execInsertSql:" + execBean.getExecSql());
        execData(execBean);
    }

    private static void updateData(List<CanalEntry.Column> columns, String tableName) {
        ExecBean execBean = updateSql(columns, tableName);
        logger.info("execUpdateSql:" + execBean.getExecSql());
        execData(execBean);
    }

    private static void deleteData(List<CanalEntry.Column> columns, String tableName) {
        ExecBean execBean = deleteSql(columns, tableName);
        logger.info("execDeleteSql:" + execBean.getExecSql());
        execData(execBean);
    }

    private static void execData(ExecBean execBean) {
        String dbUrl = prop.getProperty("dbUrl");
        String dbUsername = prop.getProperty("dbUsername");
        String dbPassword = prop.getProperty("dbPassword");

        String[] connections = StringUtils.split(dbUrl, ",");
        for (String connection : connections) {
            Connection con = getConnection(connection, dbUsername, dbPassword);
            try {
                PreparedStatement pt = con.prepareStatement(execBean.getExecSql());
                List<ColumnBean> columnBeanList = execBean.getColumnBeanList();
                for (int i = 1; i <= columnBeanList.size(); i++) {
                    ColumnBean columnBean = columnBeanList.get(i - 1);
                    if (columnBean.getIsNull()) {
                        pt.setNull(i, columnBean.getJdbcType());
                    } else {
                        pt.setObject(i, columnBean.getColumnValue(), columnBean.getJdbcType());
                    }
                }
                pt.executeUpdate();
                pt.close();
                con.close();
            } catch (SQLException e) {
                logger.error("execSql: " + execBean.getExecSql(), e);
            }
        }
    }

    private static Connection getConnection(String url, String username, String password) {
        Connection con = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            con = DriverManager.getConnection(url, username, password);
        } catch (ClassNotFoundException | SQLException e) {
            logger.error("create Connection is error...", e);
        }
        return con;
    }

    private static void startCanal() {
        String canalUrl = prop.getProperty("canalUrl");
        int canalPort = Integer.parseInt(prop.getProperty("canalPort"));
        String subscribe = prop.getProperty("canalTable");

        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalUrl, canalPort), "example", "", "");
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(subscribe);
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                System.out.println("waiting...");
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    execEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
            }
        } finally {
            connector.disconnect();
        }

    }

    public static void main(String[] args) {
        startCanal();
    }

}
